package com.shujia.wyh.mr.kqzl;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class KQZLMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
    private List<String> list;

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, LongWritable>.Context context) throws IOException, InterruptedException {
        FileSplit inputSplit = (FileSplit) context.getInputSplit();
        String name = inputSplit.getPath().getName();


        //将一行数据转变成java中String类型
        String line = value.toString();
        //根据逗号进行切分
        String[] strings = line.split(",");

        //判断读取的是否是标题
        if("date".equals(strings[0])){
            list = new ArrayList<String>();
            //从第四列开始遍历，得到每个监测点的编号
            for(int i=3;i<strings.length;i++){
                String id = strings[i].substring(0,strings[i].length()-1);
                list.add(id);
            }
        }

        String date = strings[0];
        //取出当天的日期，检测的类型是PM2.5，并且长度是大于等于4
        if (strings.length >= 4 && "PM2.5".equals(strings[2])) {
            //日期:20180119-监测点编号:1001
            for (int i = 3, j = 0; i < strings.length; i++, j++) {
                //有些监测点未检测到PM2.5的值，可能是"",也可能是" ",也可能是null
                if("".equals(strings[i]) || " ".equals(strings[i]) || strings[i]==null){
                    strings[i] = "0";
                }

                context.write(new Text("日期:"+date+"-监测点编号==:"+list.get(j)+"$"+name),new LongWritable(Long.parseLong(strings[i])));
            }
        }
    }
}
